#include <unistd.h>
#include <mosquitto.h>
#include "service.h"
#include "z_logger.h"
#include "model.h"
#include "aliot.h"
#include "aiot_subdev_api.h"
#include "subdev_private.h"
#include "cJSON.h"
#include "sqlite.h"
#include "channel.h"
#include "../mqtt/mqtt.h"
#include "value.h"
#include "z_memory.h"
#include "redis.h"

static segment_e segments[] = {seg_do, seg_di, seg_ao, seg_ai};
static char log_file[NAME_MAX] = "/tmp/timestamp.log";

void _subdev_recv_handler(void *handle, const aiot_subdev_recv_t *packet, void *user_data) {
    switch (packet->type) {
        case AIOT_SUBDEVRECV_TOPO_ADD_REPLY:
        case AIOT_SUBDEVRECV_TOPO_DELETE_REPLY:
        case AIOT_SUBDEVRECV_TOPO_GET_REPLY:
        case AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY:
        case AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY:
        case AIOT_SUBDEVRECV_SUB_REGISTER_REPLY:
        case AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY:
            log_trace("msg_id=%d, code=%d, product=%s, device=%s, message=%s, data=%s",
                      packet->data.generic_reply.msg_id,
                      packet->data.generic_reply.code,
                      packet->data.generic_reply.product_key,
                      packet->data.generic_reply.device_name,
                      packet->data.generic_reply.message,
                      packet->data.generic_reply.data);
            break;
        case AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY:
            log_trace("msg_id=%d, product=%s, device=%s, params=%s",
                      packet->data.generic_notify.msg_id,
                      packet->data.generic_notify.product_key,
                      packet->data.generic_notify.device_name,
                      packet->data.generic_notify.params);
            break;
        default:
            break;
    }
}

void *_subdev_service_init(aliot_service_args_t *args) {
    void *subdev_handle = aiot_subdev_init();

    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_MQTT_HANDLE, args->mqtt_handle);
    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_RECV_HANDLER, _subdev_recv_handler);
    aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_USERDATA, args->seqiot);

    return subdev_handle;
}

typedef struct {
    char product_key[MAX_INPUT];
    char product_secret[MAX_INPUT];
    char device_key[MAX_INPUT];
    char device_secret[MAX_INPUT];
    long timestamp;
    bool offline_tag;
} subdev_info_t;

subdev_info_t *subdev_info_new() {
    return z_calloc(1, sizeof(subdev_info_t));
}

void subdev_info_free(subdev_info_t *ptr) {
    if (!ptr) return;

    z_free(ptr);
}

static void _subdev_service_fini(void **handle) {
    aiot_subdev_deinit(handle);
}

static int mosquitto_check_rc(int rc, const char *method) {
    if (MOSQ_ERR_ERRNO == rc || MOSQ_ERR_KEEPALIVE == rc) {
        log_error("%s = %d, %s", method, errno, strerror(errno));
    } else if (MOSQ_ERR_SUCCESS != rc) {
        log_error("%s = %d, %s", method, rc, mosquitto_strerror(rc));
    }
    return rc;
}

static void _subdev_register(void *handle, subdev_info_t *subdev_info) {
    aiot_subdev_dev_t devs;
    devs.device_name = subdev_info->device_key;
    devs.device_secret = subdev_info->device_secret;
    devs.product_key = subdev_info->product_key;
    devs.product_secret = subdev_info->product_secret;

    int32_t res;

    res = aiot_subdev_send_topo_add(handle, &devs, 1);
    aliot_error_check("aiot_subdev_send_topo_add", res);

    res = aiot_subdev_send_batch_login(handle, &devs, 1);
    aliot_error_check("aiot_subdev_send_batch_login", res);
}

static void _subdev_unregister(void *handle, subdev_info_t *subdev_info) {
    aiot_subdev_dev_t devs;
    devs.device_name = subdev_info->device_key;
    devs.device_secret = subdev_info->device_secret;
    devs.product_key = subdev_info->product_key;
    devs.product_secret = subdev_info->product_secret;
    int32_t res;
    res = aiot_subdev_send_batch_logout(handle, &devs, 1);
    aliot_error_check("aiot_subdev_send_batch_logout", res);

}

typedef struct {
    void *handle;
    subdev_info_t *subdev_info_bak[MAX_INPUT];
    int num;
} user_data_t;

user_data_t *new_user_data() {
    return z_calloc(1, sizeof(user_data_t));
};

void user_data_free(user_data_t *ptr) {
    if (!ptr) return;

    z_free(ptr);
}

static void judge_offline(user_data_t *userData, device_t *device, value_t *value) {
    time_t cur_time = time(NULL);
    for (int i = 0; i < userData->num; i++) {
        if (!userData->subdev_info_bak[i]->offline_tag &&
            device->device_key == userData->subdev_info_bak[i]->device_key) {
            //比较redis和当下的时间戳，超过30min未接收
            double diff_time = difftime(cur_time, value->timestamp);
            if (diff_time > 30 * 60) {
                userData->subdev_info_bak[i]->offline_tag = true;
                _subdev_unregister(userData->handle, userData->subdev_info_bak[i]);
            }
        }
    }
}

static void aiot_login_message(user_data_t *user_data, device_t *device, value_t *value) {
    int flag = 1;//未login
    for (int i = 0; i < user_data->num; i++) {
        if (0 == strcmp(user_data->subdev_info_bak[i]->device_key, device->device_key)) {
            if (user_data->subdev_info_bak[i]->offline_tag) {
                flag = -1;//已login但未上线
                user_data->subdev_info_bak[i]->timestamp = value->timestamp;
            } else {
                flag = 0;//已login且在线
                //fixme 需要修改，目前根据时间强行判断(设备若保持链接每隔8个小时，强制上线一次)
                if (value->timestamp - user_data->subdev_info_bak[i]->timestamp >= 60 * 8) {
                    flag = -1;
                    user_data->subdev_info_bak[i]->timestamp = value->timestamp;
                }
            }
            user_data->subdev_info_bak[i]->offline_tag = false;
        }
    }
    if (flag > 0) {
        log_trace("device_key %s start to add top and login!", device->device_key);
        subdev_info_t *subdev_info = subdev_info_new();
        subdev_handle_t *sub_handler = user_data->handle;
        sprintf(subdev_info->device_key, "%s", device->device_key);
        sprintf(subdev_info->device_secret, "%s", device->device_secret);
        sprintf(subdev_info->product_key, "%s", device->_product->product_key);
        sprintf(subdev_info->product_secret, "%s", device->_product->product_secret);
        subdev_info->timestamp = value->timestamp;
        _subdev_register(user_data->handle, subdev_info);
        char topic[MAX_INPUT];
        sprintf(topic, "/%s/%s/user/get", device->_product->product_key, device->device_key);
        int res = aiot_mqtt_sub(sub_handler->mqtt_handle, topic, NULL, 1, NULL);
        if (res < 0) {
            log_error("aiot_mqtt_sub failed, res: -0x%04X\n", -res);
        }
        subdev_info->offline_tag = 0;
        user_data->subdev_info_bak[user_data->num] = subdev_info;
        user_data->num++;
    } else if (flag < 0) {
        log_trace("device_key %s start to login!", device->device_key);
        aiot_subdev_dev_t devs;
        devs.device_name = device->device_key;
        devs.device_secret = device->device_secret;
        devs.product_key = device->_product->product_key;
        devs.product_secret = device->_product->product_secret;
        int32_t res;
        res = aiot_subdev_send_batch_login(user_data->handle, &devs, 1);
        aliot_error_check("aiot_subdev_send_batch_login", res);
    }
}

static int aiot_send_message(user_data_t *user_data, device_t *device, value_t *value, thing_model_t *tag) {
    if (strlen(tag->metric) == 0) {
        return -1;
    }
    char topic[MAX_INPUT];
    int res;
    subdev_handle_t *sub_handler = user_data->handle;
    sprintf(topic, "/sys/%s/%s/thing/event/property/post", device->_product->product_key, device->device_key);
    cJSON *metric_json = cJSON_CreateObject();
    cJSON *params_json = cJSON_CreateObject();
    cJSON *payload_json = cJSON_CreateObject();
    char metric[MAX_INPUT];
    sprintf(metric, "%s", tag->metric);
    cJSON_AddNumberToObject(metric_json, "value", value->value);
    cJSON_AddItemToObject(params_json, metric, metric_json);
    cJSON_AddItemToObject(payload_json, "params", params_json);
    char *payload = cJSON_PrintUnformatted(payload_json);
    res = aiot_mqtt_pub(sub_handler->mqtt_handle, topic, (uint8_t *) payload, strlen(payload), 0);
    if (res < 0) {
        log_error("aiot_mqtt_pub failed, res: -0x%04X\n", -res);
        return -1;
    } else {
        log_trace("dev_id=%d, metric=%s, value=%f", device->super.id, metric, value->value);
        FILE *fp = fopen(log_file, "a+");
        fseek(fp, 0L, SEEK_END);
        long size = ftell(fp);
        //超过长度4K重写，将块固定为4K
        if (size >= PATH_MAX) {
            fclose(fp);
            fp = fopen(log_file, "w+");
        }
        fwrite(&value->timestamp, sizeof(time_t), 1, fp);
        fclose(fp);
    }
    return 0;
}

static int login_or_send(user_data_t *user_data, device_t *device, value_t *value, thing_model_t *tag) {
    //设备上线
    aiot_login_message(user_data, device, value);
    time_t t = time(NULL);
    //设备物模型上报
    if (value->_changed && 0 == t % 2) {
        value->_changed = false;
        return aiot_send_message(user_data, device, value, tag);
    }
    return 0;
}

static void subdev_send_message(user_data_t *user_data, device_t *device) {
    int i, j;
    for (i = 0; i < sizeof(segments) / sizeof(segments[0]); ++i) {
        segment_e segment = segments[i];
        vector_t *tag_list = &device->_product->thing_model[segment];
        vector_t *val_list = &device->thing_value[segment];

        for (j = 0; j < vector_size(val_list); ++j) {
            thing_model_t *tag = vector_get(tag_list, j);
            value_t *value = vector_get(val_list, j);
            //判断是否离线
            judge_offline(user_data, device, value);
            if (value->timestamp >= TIME_20100101) {
                int res;
                res = login_or_send(user_data, device, value, tag);
                if (res < 0)
                    log_error("login_or_send error!");
            }
        }
    }
}

//传输子设备物模型信息
static void subdev_pub(user_data_t *user_data, seqiot_t *seqiot) {
    int i_cat, i_chn, i_dev;

    struct timeval _tv;
    gettimeofday(&_tv, NULL);

    //扫描所有的数据变更
    for (i_cat = 0; i_cat < vector_size(&seqiot->catalogs); ++i_cat) {
        data_catalog_t *catalog = vector_get(&seqiot->catalogs, i_cat);

        if (catalog->_next_sched && catalog->_next_sched <= _tv.tv_sec) {
            catalog->_next_sched = next_sched(_tv.tv_sec, &catalog->pub);

            struct tm _tm;
            localtime_r(&catalog->_next_sched, &_tm);
            for (i_chn = 0; i_chn < vector_size(&seqiot->channels); ++i_chn) {
                channel_t *channel = vector_get(&seqiot->channels, i_chn);
                //从redis获取实时数据
                sched_mqtt_channel(user_data->handle, catalog->catalog, channel);
            }

            //发送数据
            for (i_chn = 0; i_chn < vector_size(&seqiot->channels); ++i_chn) {
                channel_t *channel = vector_get(&seqiot->channels, i_chn);

                for (i_dev = 0; i_dev < vector_size(&channel->devices); ++i_dev) {
                    device_t *device = vector_get(&channel->devices, i_dev);
                    subdev_send_message(user_data, device);
                }
            }
        }
    }
}

static void *_subdev_service_start(void *handle) {
    seqiot_t *seqiot = seqiot_new();
    read_config(seqiot, NULL, 0, 0);
    struct timeval _tv;
    gettimeofday(&_tv, NULL);
    //必须调用一次__timezone才有值
    localtime(&_tv.tv_sec);
    for (int i = 0; i < vector_size(&seqiot->catalogs); ++i) {
        data_catalog_t *catalog = vector_get(&seqiot->catalogs, i);
        if (catalog->pub.period < 0) {
            catalog->_next_sched = _tv.tv_sec;
        } else {
            catalog->_next_sched = next_sched(_tv.tv_sec, &catalog->pub);
        }
    }
    if (!redis_open(seqiot->local_address)) {
        return NULL;
    }

    subdev_handle_t *subdev_handle = handle;
    user_data_t *user_data = new_user_data();
    user_data->handle = subdev_handle;
    user_data->num = 0;
    while (running) {
        subdev_pub(user_data, seqiot);
        usleep(1000 * 1000);
    }
    redis_close();
    seqiot_free(seqiot);

    for (int i = 0; i < user_data->num; i++) {
        subdev_info_free(user_data->subdev_info_bak[i]);
    }
    user_data_free(user_data);
    return NULL;
}

static aliot_service_t _subdev_service = {
        .name = SUBDEV_MODULE_NAME,
        .init = _subdev_service_init,
        .fini = _subdev_service_fini,
        .start = _subdev_service_start,
        .stop = NULL,
};

aliot_service_t *subdev_service_get() {
    return &_subdev_service;
}


